package com.zgw.zookeep;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.commons.collections4.map.HashedMap;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.stereotype.Component;

import com.alibaba.dubbo.common.utils.CollectionUtils;
import com.zgw.constant.DubboConstant;

import lombok.extern.slf4j.Slf4j;
@Slf4j
@Component
public class ZookeepUtils {

	@Autowired
	ConfigurableApplicationContext context;
	
//	@Value("${host}")
	private String host;

//	@Value(("${server.port}"))
//	private String port;

//	@Value(("${spring.application.name}"))
	private String applicationName;

	static Watcher watcher;
    
	static ZooKeeper zk;
	
	static {
		if(DubboConstant.isZkDev) {
			watcher = new Watcher() {
				// 事件通知
				@Override
				public void process(WatchedEvent event) {
					// 事件状态
					KeeperState state = event.getState();
					// 连接状态
					if (KeeperState.SyncConnected == state) {
						EventType type = event.getType();
						// 事件类型
						if (EventType.None == type) {
							System.out.println("--------zk开始连接.");
						}
					}
				}
			};
			final String address = "127.0.0.1:2181";
			final int session_timeout = 5000;
			try {
				zk= new ZooKeeper(address, session_timeout, watcher);
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
 
	
	/**
	 * 服务器启动获取配置注册到zk上
	 * @throws IOException
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	@PostConstruct
	public void registerIP() throws IOException, KeeperException, InterruptedException {
		if(DubboConstant.isZkDev) {
			if(checkUrlIsExist()) {
				ZooKeeper zk = getZookeepCon();
				if (nodeIsExit(zk)) {
					String message = getNodeData(zk);
					if (message.contains("&")) {
						String[] urls = message.split("&");
						for (String item : urls) {
							if (item.equals(getUrl())) {
								return;
							}
						}
						String newMessage="";
						if(StringUtils.isNotBlank(message)) {
							newMessage = message + "&" + getUrl();
						}else
							newMessage= getUrl();
						
						setZkData(zk, settleData(newMessage));
					} else {
						if (!message.equals(getUrl())) {
							setZkData(zk, getUrl()+ "&" );
						}
					}
				} else {
					zk.create("/Project-Dubbo", getUrl().getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				}
			}
		}
	}

	private String settleData(String newMessage) {
		return Arrays.asList(newMessage.split("&"))
				.stream()
		        .collect(Collectors.toSet())
		        .stream()
		        .collect(Collectors.joining("&"));
	}

	/**
	 * 服务器关闭吧自己信息在zk上注销
	 * @throws IOException
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	@PreDestroy
	public void destory() throws IOException, KeeperException, InterruptedException {
		if(DubboConstant.isZkDev) {
			String url = getUrl();
			ZooKeeper zk = getZookeepCon();
			if (nodeIsExit(zk)) {
				String message = getNodeData(zk);
				if (message.contains("&")) {
					String[] urls = message.split("&");
					String newMessage = "";
					for (String item : urls) {
						if (!item.equals(url)) {
							newMessage += "&" + item;
						}
					}
					setZkData(zk, newMessage);
				} else {
					if (url.equals(message)) {
						setZkData(zk, "");
					}
				}
			}
		}
	}

	/**
	 * zk给指定节点赋值
	 * @param zk
	 * @param newMessage
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public void setZkData(ZooKeeper zk, String newMessage) throws KeeperException, InterruptedException {
		zk.setData("/Project-Dubbo", newMessage.getBytes(), -1);
	}

	/**
	 * 获取zk指定节点数据
	 * @param zk
	 * @return
	 * @throws UnsupportedEncodingException
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public String getNodeData(ZooKeeper zk) throws UnsupportedEncodingException, KeeperException, InterruptedException {
		Stat stat = new Stat();
		return new String(zk.getData("/Project-Dubbo", true, stat), "UTF-8");
	}

	/**
	 * 获取zk连接
	 * @return
	 * @throws IOException
	 */
	public ZooKeeper getZookeepCon() throws IOException {
//		final String address = "127.0.0.1:2181";
//		final int session_timeout = 5000;
//		return new ZooKeeper(address, session_timeout, watcher);
		return zk;
	}

	/**
	 * 判断zk是否存在自定义网络节点
	 * @param zk
	 * @return
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public boolean nodeIsExit(ZooKeeper zk) throws KeeperException, InterruptedException {
		if (zk != null) {
			Stat exists = zk.exists("/Project-Dubbo", watcher);
			if (exists != null) {
				return true;
			} else
				return false;
		}
		return false;
	}

	/**
	 * 获取服务器当前的网络地址
	 * @return
	 */
	public String getUrl() {
		return "http://" + host + "/"  + "," + applicationName;
	}
	
	/**
	 * 删除zk上的节点
	 * @param list
	 * @param zk
	 * @throws UnsupportedEncodingException
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public void removeNode(List<String> list,ZooKeeper zk ) throws UnsupportedEncodingException, KeeperException, InterruptedException {
		if(DubboConstant.isZkDev) {
			if(CollectionUtils.isNotEmpty(list)) {
				String nodeData = getNodeData(zk);
				if(StringUtils.isNotBlank(nodeData)) {
					if(nodeData.contains("&")) {
						List<String> finalIps=Arrays.asList(nodeData.split("&"));
						List<String>ips=new ArrayList<String>(finalIps);
						list.forEach((v)->{
							for(String item:finalIps) {
								if(item.equals(v)) {
									ips.remove(item);
								}
							}
						});
						if(CollectionUtils.isNotEmpty(ips)) {
							this.setZkData(zk, ips.stream().collect(Collectors.joining("&")));
						}else
							this.setZkData(zk, "");
					}else {
						if(!Optional.ofNullable(list
								.stream()
								.filter((v)->v.equals(nodeData)))
								.isPresent()) {
							this.setZkData(zk, "");
						}
					}
				}
			}
		}
	}
	
	public boolean checkUrlIsExist() {
		ConfigurableEnvironment environment = context.getEnvironment();
		String host = environment.getProperty("host");
//		String port = environment.getProperty("server.port");
		String applicationName = environment.getProperty("spring.application.name");
//		if(StringUtils.isAnyBlank(host,port,applicationName)) {
		if(StringUtils.isAnyBlank(host,applicationName)) {
			return false;
		}else {
			this.host=host;
//			this.port=port;
			this.applicationName=applicationName;
			return true;
		}
	}
	
	
    public Map<String,List<String>> getZkServerDetail(Map<String,String> gateWayMapping) throws Exception {
       if(DubboConstant.isZkDev) {
    		ZooKeeper zk=getZookeepCon();
        	String nodeData = this.getNodeData(zk);
            if(StringUtils.isNotBlank(nodeData)) {
            	String[] split = nodeData.split("&");
            	Map<String,List<String>> map=new HashedMap<String, List<String>>();
            	for(String item :split) {
            		if(StringUtils.isNotBlank(item) 
            				& item.contains(",")) {
            			String[] url = item.split(",");
            			String value=url[0];
            			String serverName=url[1];
            			map.compute(serverName, (k,v)->{
            				if(! CollectionUtils.isEmpty(v)) {
            					v.add(value);
            					return v;
            				}else {
            					List<String> list=new ArrayList<String>();
            					list.add(value);
            					return list;
            				}
            			});
            		}
            	} 
                HashedMap<String, List<String>> returnMap=new HashedMap<String, List<String>>();
                gateWayMapping.forEach((k,v)->{
                	if( ! map.isEmpty()) {
                		map.forEach((k2,v2)->{
                			if(v.equals(k2)) {
                				if(! returnMap.containsKey(k)) {
                					returnMap.put(k,v2);
                				}
                			}
                		});
                	}
                });
                return returnMap;
            }else
            	return null;
       }
       return null;
    }
    
}
